package io.kiki.sba.registry.server.metadata.metadataServer.impl;


import io.kiki.sba.registry.common.model.metaserver.cluster.VersionedList;
import io.kiki.sba.registry.common.model.metaserver.nodes.CentralControlNode;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.server.metadata.lease.dataStoreNode.DataStoreNodeManager;
import io.kiki.sba.registry.server.metadata.lease.session.ClientInterfaceNodeManager;
import io.kiki.sba.registry.server.metadata.metadataServer.CurrentNodeClusterMetadataServer;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;


public class LocalMetadataServer extends AbstractMetadataServer implements CurrentNodeClusterMetadataServer {
    private static final Logger logger = LoggerFactory.getLogger(LocalMetadataServer.class);
    private final AtomicLong currentEpoch = new AtomicLong();


    protected SlotManager slotManager;


    protected DataStoreNodeManager dataStoreNodeManager;


    protected ClientInterfaceNodeManager clientInterfaceNodeManager;

    public LocalMetadataServer() {
    }

    public LocalMetadataServer(SlotManager slotManager, DataStoreNodeManager dataStoreNodeManager, ClientInterfaceNodeManager clientInterfaceNodeManager) {
        this.slotManager = slotManager;
        this.dataStoreNodeManager = dataStoreNodeManager;
        this.clientInterfaceNodeManager = clientInterfaceNodeManager;
    }

    @Override
    public Slots getSlots() {
        return slotManager.getSlots();
    }

    public long getEpoch() {
        return currentEpoch.get();
    }

    @Override
    public List<CentralControlNode> getList() {
        return new ArrayList<>(centralControlNodeSet);
    }

    @Override
    public void updateClusterMemberList(VersionedList<CentralControlNode> list) {
        if (list.getEpoch() <= currentEpoch.get()) {
            logger.warn("[updateClusterMembers]Epoch[{}] is less than current[{}], ignore: {}", currentEpoch.get(), list.getEpoch(), list.getList());
        }
        lock.writeLock().lock();
        try {
            logger.warn("[updateClusterMembers] update meta-servers, \nprevious[{}]: {} \ncurrent[{}]: {}", currentEpoch.get(), getList(), list.getEpoch(), list.getList());
            currentEpoch.set(list.getEpoch());
            centralControlNodeSet = new HashSet<>(list.getList());
        } finally {
            lock.writeLock().unlock();
        }
    }

    @Override
    public DataStoreNodeManager getDataStoreNodeManager() {
        return dataStoreNodeManager;
    }

    @Override
    public ClientInterfaceNodeManager getClientInterfaceNodeManager() {
        return clientInterfaceNodeManager;
    }

    @Override
    public void renew(CentralControlNode centralControlNode) {
        lock.writeLock().lock();
        try {
            logger.info("[renew]meta node [{}] renewed", centralControlNode);
            centralControlNodeSet.add(centralControlNode);
        } finally {
            lock.writeLock().unlock();
        }
    }

    @Override
    public void cancel(CentralControlNode centralControlNode) {
        lock.writeLock().lock();
        try {
            logger.info("[cancel]meta node [{}] removed", centralControlNode);
            centralControlNodeSet.remove(centralControlNode);
        } finally {
            lock.writeLock().unlock();
        }
    }

    @Override
    public VersionedList<CentralControlNode> getCluster() {
        return new VersionedList<>(getList(), getEpoch());
    }
}
